package secsmarts;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Locale;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import secsmarts.LoadFileToHbase4.Record;

public class LoadFileToHbase3 extends Configured implements Tool {
	public static class MapperClass extends Mapper<LongWritable, Text, Record, Text> {
		// 列名
		public static final String[] COLUMNS = { "time", "url", "ip", "country", "code", "type" };

		@Override
		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			String[] cloumnVals = value.toString().split("\t");
			if (cloumnVals.length == 6) {

				String time = "";
				try {
					time = new SimpleDateFormat("yyyyMMddHHmmss")
							.format(new SimpleDateFormat("d/M/yyyy h:mm:ss a", Locale.ENGLISH).parse(cloumnVals[0]));
				} catch (ParseException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}

				Record r = new Record(time, cloumnVals[1], cloumnVals[2], cloumnVals[3], cloumnVals[4], cloumnVals[5]);

				context.write(r, new Text(""));
			}
		}

	}

	public static class ReduceClass extends TableReducer<Record, Text, ImmutableBytesWritable> {
		private String gethost(String url) {
			// TODO Auto-generated method stub
			if (url != null && url.startsWith("http://")) {
				String tmp = url.substring(7);
				return url.substring(0, 7 + tmp.indexOf("/"));
			}
			return null;
		}

		@Override
		public void reduce(Record r, Iterable<Text> values, Context context) throws IOException, InterruptedException {
			// public void reduce(Record r, Iterator<Text> values, Context
			// context) throws IOException , InterruptedException {
			// TODO Auto-generated method stub
			String url = r.url.toString(), ip = r.ip.toString(), host = gethost(url);
			String[] rowkeys = new String[] { url, ip, host };
			for (String rowkey : rowkeys) {
				if (rowkey != null) {
					rowkey = String.format("%03d", Math.abs(rowkey.hashCode()) % REGIONS) + rowkey;
					Put put = new Put(rowkey.getBytes());

					put.addColumn("info".getBytes(), "country".getBytes(), r.getCountry().getBytes());
					put.addColumn("info".getBytes(), "time".getBytes(), r.getTime().getBytes());
					put.addColumn("info".getBytes(), "ip".getBytes(), r.getIp().getBytes());
					put.addColumn("info".getBytes(), "code".getBytes(), r.getCode().getBytes());
					put.addColumn("info".getBytes(), "url".getBytes(), r.getUrl().getBytes());
					put.addColumn("info".getBytes(), "type".getBytes(), r.getType().getBytes());

					context.write(new ImmutableBytesWritable(rowkey.getBytes()), put);
				}
			}
		}
	}

	static class GroupComparator extends WritableComparator {
		protected GroupComparator() {
			super(Record.class, true);
		}

		@Override
		public int compare(WritableComparable a, WritableComparable b) {
			Record r1 = (Record) a;
			Record r2 = (Record) b;
			return r2.time.compareTo(r1.time);
		}
	}

	public static class SortPartitioner extends Partitioner<Record, Text> {
		public SortPartitioner() {
			// TODO Auto-generated constructor stub
		}

		@Override
		public int getPartition(Record text, Text value, int numReducers) {
			return Math.abs(text.getUrl().toString().hashCode()) % numReducers;
		}
	}

	public static class Record implements WritableComparable<Record> {
		Text time = new Text(), url = new Text(), ip = new Text(), country = new Text(), code = new Text(),
				type = new Text();

		public Record() {
		}

		public Record(String time, String url, String ip, String country, String code, String type) {
			super();
			if (time != null)
				this.time = new Text(time);
			if (url != null)
				this.url = new Text(url);
			if (ip != null)
				this.ip = new Text(ip);
			if (country != null)
				this.country = new Text(country);
			if (code != null)
				this.code = new Text(code);
			if (type != null)
				this.type = new Text(type);
		}

		@Override
		public void readFields(DataInput in) throws IOException {
			// TODO Auto-generated method stub
			// time = arg0.readUTF();
			// url = arg0.readUTF();
			// ip = arg0.readUTF();
			// country = arg0.readUTF();
			// code = arg0.readUTF();
			// type = arg0.readUTF();
			time.readFields(in);
			url.readFields(in);
			ip.readFields(in);
			country.readFields(in);
			code.readFields(in);
			type.readFields(in);

		}

		@Override
		public void write(DataOutput out) throws IOException {
			// TODO Auto-generated method stub
			// arg0.writeUTF(time);
			// arg0.writeUTF(url);
			// arg0.writeUTF(ip);
			// arg0.writeUTF(country);
			// arg0.writeUTF(code);
			// arg0.writeUTF(type);
			time.write(out);
			url.write(out);
			ip.write(out);
			country.write(out);
			code.write(out);
			type.write(out);

		}

		public int compareTo(Record o) {
			// TODO Auto-generated method stubss
			return url.compareTo(o.getUrl());
		}

		public Text getTime() {
			return time;
		}

		public void setTime(Text time) {
			this.time = time;
		}

		public Text getUrl() {
			return url;
		}

		public void setUrl(Text url) {
			this.url = url;
		}

		public Text getIp() {
			return ip;
		}

		public void setIp(Text ip) {
			this.ip = ip;
		}

		public Text getCountry() {
			return country;
		}

		public void setCountry(Text country) {
			this.country = country;
		}

		public Text getCode() {
			return code;
		}

		public void setCode(Text code) {
			this.code = code;
		}

		public Text getType() {
			return type;
		}

		public void setType(Text type) {
			this.type = type;
		}

		@Override
		public String toString() {
			return "Record [time=" + time + ", url=" + url + ", ip=" + ip + ", country=" + country + ", code=" + code
					+ ", type=" + type + "]";
		}

	}

	static String tableName = "bingmurls1";

	public static void main(String[] args) {

		int exitCode = -1;
		try {
			exitCode = ToolRunner.run(new LoadFileToHbase3(), args);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		System.exit(exitCode);
	}

	private static void createTable() throws IOException {
		// TODO Auto-generated method stub

		Configuration conf = HBaseConfiguration.create();
		conf.set("hbase.zookeeper.quorum", "slave01.secsmarts.com,master.secsmarts.com,slave02.secsmarts.com");
		conf.set("zookeeper.znode.parent", "/hbase-unsecure");
		HBaseAdmin admin = new HBaseAdmin(conf);
		if (admin.tableExists(tableName)) {
			return;
		}
		System.out.println("table not exists!creating.......");
		// admin.disableTable(tableName);
		// admin.deleteTable(tableName);
		HTableDescriptor htd = new HTableDescriptor(tableName);
		HColumnDescriptor tcd = new HColumnDescriptor("info");
		htd.addFamily(tcd);// 创建列族
		byte[][] splits = new byte[REGIONS][3];
		for (int i = 0; i < REGIONS; i++) {
			splits[i] = String.format("%03d", i).getBytes();
		}
		admin.createTable(htd, splits);// 创建表
	}

	public static int REGIONS = 100;

	public int run(String[] args) throws Exception {
		// TODO Auto-generated method stub

		createTable();
		Configuration conf = getConf();
		conf.set("fs.defaultFS", "hdfs://secsmarts");
		conf.set("mapreduce.framework.name", "yarn");
		// conf.set("mapred.job.tracker", "ubuntu:9001");
		conf.set("hbase.zookeeper.quorum", "slave01.secsmarts.com,master.secsmarts.com,slave02.secsmarts.com");
		conf.set("hbase.zookeeper.property.clientPort", "2181");
		conf.set("zookeeper.znode.parent", "/hbase-unsecure");

		Job job = Job.getInstance(conf);
		job.setJarByClass(LoadFileToHbase3.class);
		job.setMapperClass(MapperClass.class);
		job.setMapOutputKeyClass(Record.class);
		job.setMapOutputValueClass(Text.class);
		job.setGroupingComparatorClass(GroupComparator.class);
		job.setReducerClass(ReduceClass.class);
		job.setPartitionerClass(SortPartitioner.class);
		job.setOutputFormatClass(TableOutputFormat.class);
		job.setOutputKeyClass(ImmutableBytesWritable.class);
		job.setOutputValueClass(Put.class);
		TableMapReduceUtil.initTableReducerJob(tableName, ReduceClass.class, job);

		FileInputFormat.addInputPath(job, new Path(args[0]));

		return job.waitForCompletion(true) ? 0 : 1;

	}
}
